package com.huawei.app.demo.video;

import java.io.Closeable;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.ArrayBlockingQueue;

import com.huawei.app.demo.util.LogUtils;
import com.huawei.app.httpapi.ThreadPools;

/**
 * 数据发送任务
 *
 * @since 2020-08-14
 */
public class SenderTask implements Runnable {

    private final static String TAG = SenderTask.class.getSimpleName();

    private String host;
    private int port;
    private boolean isRuning = true;
    private boolean connectOk = false;

    private Socket socket;
    private OutputStream outputStream;
    private ObjectOutputStream objOutputStream;
    private ArrayBlockingQueue<ImgBuffer> sendQueue;

    public SenderTask(String host, int port, ArrayBlockingQueue<ImgBuffer> sendQueue) {
        this.host = host;
        this.port = port;
        this.sendQueue = sendQueue;
    }

    private void init() {
        try
        {
            // 关闭上次资源
            close();

            // 重新发起连接
            this.socket = new Socket();
            this.socket.connect(new InetSocketAddress(host, port), 10000);
            this.outputStream = socket.getOutputStream();
            this.objOutputStream = new ObjectOutputStream(outputStream);
            this.connectOk = true;
            LogUtils.getInstance().i(TAG, "init socket client on " + host + ":" + port + " success.");
        } catch (IOException e) {
            LogUtils.getInstance().e(TAG, "init socket client on " + host + ":" + port + " failed, " + e.getMessage());
            this.connectOk = false;
        }
    }

    @Override
    public void run() {
        String oldThreadName = Thread.currentThread().getName();
        Thread.currentThread().setName(this.getClass().getSimpleName());

        ImgBuffer imgBuffer = null;
        while (isRuning) {
            try {
                if (!isConnected()) {
                    init();
                    if (!isConnected()) {
                        ThreadPools.pause(1000);
                        continue;
                    }
                }
                ThreadPools.pause(20);

                if (imgBuffer == null) {
                    LogUtils.getInstance().v(TAG, "before sendQueue.poll.");
                    imgBuffer = this.sendQueue.poll();
                    if (imgBuffer == null) {
                        ThreadPools.pause(20);
                        continue;
                    }
                }

                LogUtils.getInstance().v(TAG, "before objOutputStream.writeObject sno: " + imgBuffer.getSno() + ", snoIndex: " + imgBuffer.getSnoIndex());
                this.objOutputStream.writeObject(imgBuffer);
                LogUtils.getInstance().v(TAG, "after objOutputStream.writeObject sno: " + imgBuffer.getSno()  + ", snoIndex: " + imgBuffer.getSnoIndex());
                imgBuffer = null;
            } catch (Exception e) {
                this.connectOk = false;
                LogUtils.getInstance().e(TAG, e.getMessage());
            }
        }

        Thread.currentThread().setName(oldThreadName);
    }

    private boolean isConnected() {
        return (this.socket != null && this.socket.isConnected() && !this.socket.isClosed() && this.connectOk);
    }

    public void stop() {
        this.isRuning = false;
        close();
    }

    private void close() {
        closeQuietly(this.objOutputStream);
        closeQuietly(this.outputStream);
        closeQuietly(this.socket);
    }

    private void closeQuietly(Closeable io) {
        try {
            if (io != null) {
                io.close();
            }
        } catch (IOException e) {
            // ignore.
        }
    }
}
